package org.gorson.utils.es;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;
import org.gorson.utils.CastUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import static org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase.FlushBackoffType.CONSTANT;

/**
 * ES sink，用于新增或覆盖操作
 */
public class DefaultESIndexDataStreamSink implements Serializable {
    private final Logger logger = LoggerFactory.getLogger(DefaultESIndexDataStreamSink.class);

    /**
     * 获取还款es sink builder
     *
     * @param properties ES配置
     * @param esIndex    索引名称
     * @param esIdName   主键字段名称，注意：该字段对应的值必须是String类型
     * @return ElasticsearchSink.Builder<String>
     */
    public ElasticsearchSink.Builder<String> getBuilder(Map properties, String esIndex, String esIdName) {

        ESProperties esProperties = new ESProperties(properties);

        //1、准备es连接参数
        List<HttpHost> httpHosts = new ArrayList<>();
        String[] hostStrs = esProperties.getHostName().split(",");
        for (String host : hostStrs) {
            httpHosts.add(new HttpHost(host, esProperties.getPort(), esProperties.getSchema()));
        }

        //2、
        ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
                httpHosts,
                new ElasticsearchSinkFunction<String>() {
                    public IndexRequest createIndexRequest(String element) {
                        //2、String转JSONObject，用于获取es id字段值
                        JSONObject obj = JSON.parseObject(element);
                        //3、数据写入es
                        return Requests.indexRequest()
                                .index(esIndex)
                                .id(CastUtil.castString(obj.get(esIdName)))
                                .source(element, XContentType.JSON);
                    }

                    @Override
                    public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
                        indexer.add(createIndexRequest(element));
                    }
                }
        );

        /*
        //默认1000。每个Bulk请求，最大缓冲Action个数
        esSinkBuilder.setBulkFlushMaxActions(1);//每次最大插入数量
        //默认5mb。每个Bulk请求，最大缓冲的Action大小。
        esSinkBuilder.setBulkFlushMaxSizeMb(1);//最大插入内存
        esSinkBuilder.setBulkFlushInterval(1);//插入刷新周期，单位毫秒。
        esSinkBuilder.setFailureHandler(new RetryRejectedExecutionFailureHandler());//处理失败的Elasticsearch请求

        */
        esSinkBuilder.setBulkFlushBackoff(true);//如果其一个或多个操作由于临时原因而失败，执行带有回退延迟的重试
        esSinkBuilder.setBulkFlushBackoffType(CONSTANT);//回退延迟的类型
        esSinkBuilder.setBulkFlushBackoffDelay(500L);//重试的时间间隔
        esSinkBuilder.setBulkFlushBackoffRetries(20);//失败重试的次数

        // provide a RestClientFactory for custom configuration on the internally created REST client
        /*esSinkBuilder.setRestClientFactory(
                restClientBuilder -> {
                    restClientBuilder.setDefaultHeaders("...")
                    restClientBuilder.setMaxRetryTimeoutMillis(2)
                    restClientBuilder.setPathPrefix("...")
                    restClientBuilder.setHttpClientConfigCallback("...")
                }
        );*/


        return esSinkBuilder;
    }
}
